package com.gitee.Transformation

import com.gitee.Transformation.partition.PartitionInspector
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment

/*
  作用: 将数据均匀打撒, 可以处理数据倾斜的情况
 */
object ReblanceTest extends PartitionInspector{
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(3)
    import org.apache.flink.api.scala._
    val data: DataSet[Long] = env.generateSequence(1,100)
    val filterData: DataSet[Long] = data.filter(_ > 8)

    // 没有进行Reblance得到的结果
    val result01: DataSet[(Long, Long)] = filterData.map(new RichMapFunction[Long, (Long, Long)] {
      override def map(value: Long): (Long, Long) = {
        val index: Long = getRuntimeContext.getIndexOfThisSubtask
        (index, value)
      }
    })
    // 很明显,没有进行Reblance数字全都跑到一个TaskManager里面去了
    result01.map(x=>(x._1,1)).groupBy(0).sum(1).print()
    println("================")
    // 进行Reblance
    val rblData: DataSet[Long] = filterData.rebalance()
    val result02: DataSet[(Long, Long)] = rblData.map(new RichMapFunction[Long, (Long, Long)] {
      override def map(value: Long): (Long, Long) = {
        val index: Long = getRuntimeContext.getIndexOfThisSubtask
        (index, value)
      }
    })
    result02.map(x=>(x._1,1)).groupBy(0).sum(1).print()

    inspector(rblData)
  }

}
